package cn.itcast.streaming.sink;

import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.StringUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Author itcast
 * Date 2021/9/24 14:47
 * Desc TODO
 */
public class TripSampleToHBaseSink extends RichSinkFunction<String[]> {
    private static final Logger logger = LoggerFactory.getLogger(TripSampleToHBaseSink.class);

    //定义当前类的私有变量
    private String tableName;
    //定义创建连接
    Connection conn = null;
    //定义表操作的对象
    BufferedMutator mutator = null;

    //2.创建一个有参数-表名的构造方法
    public TripSampleToHBaseSink(String _tableName) {
        this.tableName = _tableName;
    }

    //3.重写open方法
    @Override
    public void open(Configuration parameters) throws Exception {
        //3.1 从上下文获取到全局的参数
        ParameterTool tools = (ParameterTool) getRuntimeContext()
                .getExecutionConfig()
                .getGlobalJobParameters();
        //3.2 设置hbase的配置，Zookeeper Quorum集群和端口和TableInputFormat的输入表
        String zkQuorum = tools.getRequired("zookeeper.quorum");
        String port = tools.getRequired("zookeeper.clientPort");
        org.apache.hadoop.conf.Configuration def = new org.apache.hadoop.conf.Configuration();
        def.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum);
        def.set(HConstants.ZOOKEEPER_CLIENT_PORT, port);
        def.set(TableInputFormat.INPUT_TABLE, tableName);

        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create(def);
        //3.3 通过连接工厂创建连接
        conn = ConnectionFactory.createConnection(conf);
        //todo 设置缓存对象的多大、多长时间刷新到hbase到
        BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tableName));
        params.writeBufferSize(10 * 1024 * 1024L);
        params.setWriteBufferPeriodicFlushTimeoutMs(5 * 1000L);
        //3.4 通过连接获取表对象
        try {
            mutator = conn.getBufferedMutator(params);
        } catch (Exception ex) {
            logger.error("当前获取bufferedMutator 失败：" + ex.getMessage());
        }
    }

    //5. 重写 invoke 方法，将读取的数据写入到 hbase
    @Override
    public void invoke(String[] value, Context context) throws Exception {
        //5.1 setDataSourcePut输入参数value，返回put对象
        try {
            Put put = setDataSourcePut(value);
            mutator.mutate(put);
            //5.2 指定时间内的数据强制刷写到hbase
            mutator.flush();
        } catch (Exception ex) {
            logger.error("写入到hbase失败：" + ex.getMessage());
        }
    }

    //4.重写close方法
    @Override
    public void close() throws Exception {
        //4.1 关闭hbase 表和连接资源
        if (mutator != null) mutator.close();
        if (!conn.isClosed()) conn.close();
    }

    //6. 实现 setDataSourcePut 方法

    /**
     * 每条对象生成一个 put
     * 1.表名 2.rowkey 3.列簇  4.列名和列值
     *
     * @param tripDriveArr
     * @return
     */
    private Put setDataSourcePut(String[] tripDriveArr) {
        //6.1 如何设计rowkey VIN+时间戳翻转
        String rowkey = tripDriveArr[0]+StringUtil.reverse(tripDriveArr[1]);
        //6.3 通过 rowkey 实例化 put
        Put put = new Put(Bytes.toBytes(rowkey));
        //6.2 定义列簇的名称
        String cf = "cf";
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("vin"), Bytes.toBytes(tripDriveArr[0]));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("terminalTimeStamp"), Bytes.toBytes(tripDriveArr[1]));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("soc"), Bytes.toBytes(tripDriveArr[2]));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("mileage"), Bytes.toBytes(tripDriveArr[3]));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("speed"), Bytes.toBytes(tripDriveArr[4]));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("gps"), Bytes.toBytes(tripDriveArr[5]));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("terminalTime"), Bytes.toBytes(tripDriveArr[6]));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("processTime"), Bytes.toBytes(DateUtil.getCurrentDateTime()));

        return put;
    }
}
